package com.lpj.base.id.generator.service.impl.provider;

import com.alibaba.fastjson.JSONObject;
import com.lpj.base.id.generator.service.impl.bean.IdMeta;
import com.lpj.base.id.generator.service.impl.bean.MachineIdEntity;
import com.lpj.base.id.generator.util.IpUtils;
import com.lpj.base.id.generator.util.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.StringUtils;

import java.util.Objects;
import java.util.Timer;
import java.util.TimerTask;

/**
 * redis提供动态的机器id
 *
 * @author LiPengJu
 * @date: 2020/4/19
 */
public class RedisMachineIdProvider extends TimerTask implements MachineIdProvider {
    private final Logger logger = LoggerFactory.getLogger(RedisMachineIdProvider.class);
    private static RedisMachineIdProvider instance = null;
    private final long LOCK_EXPIRE = 3000; //ms
    private final String PREFIX = "service:id:";
    private final String machineIdKey;
    private final RedisTemplate redisTemplate;
    private Long currentMachineId = -1L;

    private RedisMachineIdProvider(String machineIdKey, RedisTemplate redisTemplate) {
        if (instance != null)
            throw new RuntimeException("Cannot instantiate more than once");
        logger.info("Initialize redis to provide machine ID");
        if (!StringUtils.hasText(machineIdKey)) {
            logger.warn("It is recommended to actively configure the redis key -> service.id.machineIdKey , use default key -> RedisMachineIdProvider");
            machineIdKey = "RedisMachineIdProvider";
        }
        this.machineIdKey = PREFIX + machineIdKey;
        this.redisTemplate = redisTemplate;
        Timer timer = new Timer();
        timer.schedule(this, MODIFIED_INTERVAL, MODIFIED_INTERVAL);
    }

    public static RedisMachineIdProvider getInstance(String machineIdKey, RedisTemplate redisTemplate) {
        if (instance == null) {
            synchronized (RedisMachineIdProvider.class) {
                if (instance == null) {
                    instance = new RedisMachineIdProvider(machineIdKey, redisTemplate);
                }
            }
        }
        return instance;
    }

    public long getMachineId(IdMeta idMeta) {
        logger.info("Start getting machine ID");
        String globalLockKey = machineIdKey + "-lock";
        boolean lock = lock(globalLockKey);
        if (lock) {
            long currentTimeMillis = System.currentTimeMillis();
            try {
                for (long i = 1; i <= idMeta.getMaxMachineId(); i++) {
                    //机器id再次利用
                    MachineIdEntity machineIdEntity = get(machineIdKey, i);
                    // 能使用的机器id：
                    // 1、machineIdEntity为null
                    // 2、上一次的时间戳不等于0且比当前时间戳小
                    // 3、 机器id不再使用且上一次的时间戳比当前时间戳小
                    //不存在则使用
                    if (machineIdEntity == null) {
                        MachineIdEntity entity = new MachineIdEntity(i);
                        entity.setType(Integer.valueOf(idMeta.getExtTimeType()));
                        entity.setLastTimestamp(0L);
                        entity.setIp(IpUtils.getHostIp());
                        set(machineIdKey, entity);
                        this.currentMachineId = i;
                        logger.info(String.format("MachineId %s obtained successfully", this.currentMachineId));
                        return i;
                    } else if ((machineIdEntity.getLastTimestamp() != 0 && TimeUtils.getTime(currentTimeMillis) > TimeUtils.getTimestamp(machineIdEntity.getLastTimestamp(), machineIdEntity.getType()))
                            || (getRedisTime() - machineIdEntity.getModifiedOn() > MODIFIED_INTERVAL_CACHE
                            && currentTimeMillis - machineIdEntity.getUserTimestamp() > MODIFIED_INTERVAL_CACHE)) {
                        machineIdEntity.setLastTimestamp(0L);
                        set(machineIdKey, machineIdEntity);
                        this.currentMachineId = i;
                        logger.info(String.format("MachineId %s obtained successfully", this.currentMachineId));
                        return i;
                    }
                }
            } finally {
                if (System.currentTimeMillis() - currentTimeMillis > LOCK_EXPIRE)
                    throw new RuntimeException("Timeout holding lock");
                unlock(globalLockKey);
            }
        } else {
            throw new RuntimeException("cannot obtain the redis global lock. Check whether the redis key is the same as other business redis keys");
        }
        throw new RuntimeException("System time called back or service reboot multiple times, resulting in no machine ID available,");
    }

    @Override
    public void setBadMachineId(long machineId, long lastTimestamp) {
        logger.warn(String.format("machineId %s used service time is called back, lastTimestamp: %s, currentTimestamp: %s", machineId, lastTimestamp, TimeUtils.getTime(System.currentTimeMillis())));
        MachineIdEntity machineIdEntity = get(machineIdKey, machineId);
        if (machineIdEntity != null) {
            machineIdEntity.setLastTimestamp(lastTimestamp);
            set(machineIdKey, machineIdEntity);
        }
    }

    public MachineIdEntity get(String machineIdKey, Long machineId) {
        HashOperations hashOperations = redisTemplate.opsForHash();
        if (!hashOperations.hasKey(machineIdKey, machineId))
            return null;
        Object valueStr = hashOperations.get(machineIdKey, machineId);
        return JSONObject.parseObject(valueStr.toString(), MachineIdEntity.class);
    }

    public void set(String machineIdKey, MachineIdEntity machineIdEntity) {
        machineIdEntity.setModifiedOn(getRedisTime());
        machineIdEntity.setUserTimestamp(System.currentTimeMillis());
        HashOperations hashOperations = redisTemplate.opsForHash();
        hashOperations.put(machineIdKey, machineIdEntity.getId(), JSONObject.toJSONString(machineIdEntity));
    }

    private boolean lock(final String machineIdLockKey) {
        return (Boolean) redisTemplate.execute((RedisCallback) connection -> {
            long expireAt = System.currentTimeMillis() + LOCK_EXPIRE + 1;
            Boolean acquire = connection.setNX(machineIdLockKey.getBytes(), String.valueOf(expireAt).getBytes());
            if (acquire) {
                return acquire;
            } else {
                int failCount = 0;
                while (failCount < 5) {
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    byte[] bytes = connection.get(machineIdLockKey.getBytes());
                    if (Objects.nonNull(bytes) && bytes.length > 0) {
                        long expiredTime = Long.valueOf(new String(bytes));
                        if (expiredTime < System.currentTimeMillis()) {
                            //过期了，重新加锁
                            long restExpireAt = System.currentTimeMillis() + LOCK_EXPIRE + 1;
                            byte[] oldBytes = connection.getSet(machineIdLockKey.getBytes(), String.valueOf(restExpireAt).getBytes());
                            long oldExpiredTime = Long.valueOf(new String(oldBytes));
                            return expiredTime == oldExpiredTime;
                        }
                    } else {
                        acquire = connection.setNX(machineIdLockKey.getBytes(), String.valueOf(expireAt).getBytes());
                        if (acquire)
                            return acquire;
                    }
                    failCount++;
                    String warnMsg = String.format("You cannot obtain the redis global lock, retry the %d time", failCount);
                    logger.warn(warnMsg);
                }
                return false;
            }
        });
    }

    private void unlock(String machineIdLockKey) {
        redisTemplate.execute((RedisCallback) connection -> {
            connection.del(machineIdLockKey.getBytes());
            return true;
        });
    }

    /**
     * 获取redis服务器时间
     *
     * @return
     */
    private long getRedisTime() {
        return (Long) redisTemplate.execute((RedisCallback) connection -> connection.time());
    }

    @Override
    public void run() {
        //对redis指定key进行心跳
        MachineIdEntity machineIdEntity = get(machineIdKey, currentMachineId);
        if (machineIdEntity != null) {
            logger.info(String.format("heartbeat -> machineId %s is in use", currentMachineId));
            machineIdEntity.setLastTimestamp(0L);
            set(machineIdKey, machineIdEntity);
        }
    }
}
